package com.example.yckjbigdataservice.config;

import com.example.yckjbigdataservice.properties.KafkaProperties;
import com.example.yckjbigdataservice.utils.HiveDataSourceUtil;
import com.example.yckjbigdataservice.utils.KafkaLoginUtil;
import com.example.yckjbigdataservice.utils.SimplePartitioner;
import kafka.utils.ShutdownableThread;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.*;

@Configuration
@Slf4j
@Order(Ordered.HIGHEST_PRECEDENCE )
//@Order(51)
public class KafkaConfig {

    // Broker地址列表
    private final static String BOOTSTRAP_SERVER = "bootstrap.servers";

    // 客户端ID
    private final static String CLIENT_ID = "client.id";

    // Key序列化类
    private final static String KEY_SERIALIZER = "key.serializer";

    // Value序列化类
    private final static String VALUE_SERIALIZER = "value.serializer";

    // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
    private final static String SECURITY_PROTOCOL = "security.protocol";

    // 服务名
    private final static String SASL_KERBEROS_SERVICE_NAME = "sasl.kerberos.service.name";

    // 域名
    private final static String KERBEROS_DOMAIN_NAME = "kerberos.domain.name";

    // 分区类名
    private final static String PARTITIONER_NAME = "partitioner.class";


    // Group id
    private final static String GROUP_ID = "group.id";

    // 消息内容使用的反序列化类
    private final static String VALUE_DESERIALIZER = "value.deserializer";

    // 消息Key值使用的反序列化类
    private final static String KEY_DESERIALIZER = "key.deserializer";


    // 是否自动提交offset
    private final static String ENABLE_AUTO_COMMIT = "enable.auto.commit";

    // 自动提交offset的时间间隔
    private final static String AUTO_COMMIT_INTERVAL_MS = "auto.commit.interval.ms";

    // 会话超时时间
    private final static String SESSION_TIMEOUT_MS = "session.timeout.ms";

    // 一次请求的最大等待时间(Ms)
    private final int waitTime = 1000;

    @Value("${custom.kafka.kafka-security-mode.file.path}")
    private String kafkaSecurityModeFilePath;

    @Value("${custom.kafka.principal}")
    private String principal;

    @Value("${custom.kafka.user-keytab.file.path}")
    private String userKeytabFilePath;

    @Value("${custom.kafka.kbr5.file.path}")
    private String kbr5FilePath;

    @Value("${custom.kafka.config.path}")
    private String configPath;

    @Value("${custom.kafka.consumer.group.id}")
    private String groupId;

    @Value("${custom.kafka.consumer.topics}")
    private String topics;

    @Value("${server.port}")
    private int port;

    @Value("${spring.application.name}")
    private String appName;

    @Bean
    public KafkaProducer<String, String> kafkaProducer() throws Exception {
        if (KafkaLoginUtil.isSecurityModel(kafkaSecurityModeFilePath)) {
            try {
                log.info("授权认证开始.");
                // !!注意，安全认证时，需要用户手动修改为自己申请的机账号
                //KafkaLoginUtil.securityPrepare(principal, userKeytabFilePath, kbr5FilePath, appName+port+"kafka");
                KafkaLoginUtil.securityPrepare(principal, userKeytabFilePath, kbr5FilePath, appName+port);

                HiveDataSourceUtil.KRB5_FILE = kbr5FilePath;
                HiveDataSourceUtil.USER_KEYTAB_FILE = userKeytabFilePath;
                HiveDataSourceUtil.USER_NAME = principal;

            } catch (Exception e) {
                log.error("认证授权失败",e);
                throw e;
            }
            log.info("认证授权失败成功.");
        }
        Properties props = initProducerProperties(configPath, appName+port+"Producer");
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // acks=0 把消息发送到kafka就认为发送成功
        // acks=1 把消息发送到kafka leader分区并且写入磁盘就认为发送成功
        // acks=all 把消息发送到kafka leader分区并且leader分区的副本follower对消息进行了同步就任务发送成功
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // KafkaProducer.send() 和 partitionsFor() 方法的最长阻塞时间 ms
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 6000);
        // 批量处理的最大大小 byte
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
        // 当生产端积累的消息达到batch-size或接收到消息linger.ms后 producer会将消息发送给kafka
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
        // 生产者可用缓冲区的最大值 byte
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 每条消息最大的大小
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);
        // 客户端ID
        //props.put(ProducerConfig.CLIENT_ID_CONFIG, "client-1");
        // Key 序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // Value 序列化方式
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 消息压缩 none、lz4、gzip、snappy
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
        // 自定义分区
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SimplePartitioner.class.getName());


        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        // 开始监听 topic
        kafkaConsumer();

        return kafkaProducer;
    }

    /**
     * 监听消息
     * @throws Exception
     */
    public void kafkaConsumer() throws Exception {
        Properties props = initConsumerProperties(configPath, groupId);

        // 是否自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 自动提交offset的时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        // 批量消费最大数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        // 请求超时s
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 120000);
        // Key 反序列化类
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // Value 反序列化类
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅
        String threadName = "Kafka"+appName+port+"Consumer";
        List<String> topicsList = new ArrayList<>();
        topicsList.add(topics);
        if(topics.contains(",")){
            topicsList = Arrays.asList(topics.split(","));
        }
        //consumer.subscribe(Collections.singletonList("this.topic"));
        consumer.subscribe(topicsList);

        ShutdownableThread thread = new ShutdownableThread(threadName, false){
            @Override
            public void doWork() {
                ConsumerRecords<String, String> records = consumer.poll(Duration.of(waitTime, ChronoUnit.MINUTES));
                // 消息处理
                for (ConsumerRecord<String, String> record : records) {
                    log.info(threadName+", 收到消息: (" + record.key() + ", " + record.value()
                            + ") at offset " + record.offset());
                }
            }
        };
        thread.start();
    }


    public static Properties initProducerProperties(String configPath, String clientId) throws Exception {
        Properties props = new Properties();
        KafkaProperties kafkaProc = KafkaProperties.getInstance(configPath);
        // Broker地址列表
        props.put(BOOTSTRAP_SERVER, kafkaProc.getValues(BOOTSTRAP_SERVER, "localhost:21007"));
        // 客户端ID
        //props.put(CLIENT_ID, kafkaProc.getValues(CLIENT_ID, "DemoProducer"));
        props.put(CLIENT_ID, clientId);
        // Key序列化类
        props.put(KEY_SERIALIZER,
                kafkaProc.getValues(KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer"));
        // Value序列化类
        props.put(VALUE_SERIALIZER,
                kafkaProc.getValues(VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer"));
        // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
        props.put(SECURITY_PROTOCOL, kafkaProc.getValues(SECURITY_PROTOCOL, "SASL_PLAINTEXT"));
        // 服务名
        props.put(SASL_KERBEROS_SERVICE_NAME, "kafka");
        // 域名
        props.put(KERBEROS_DOMAIN_NAME, kafkaProc.getValues(KERBEROS_DOMAIN_NAME, "hadoop.hadoop.com"));
        // 分区类名
        /*props.put(PARTITIONER_NAME,
                kafkaProc.getValues(PARTITIONER_NAME, "com.example.yckjbigdataservice.utils.SimplePartitioner"));*/
        return props;
    }

    public static Properties initConsumerProperties(String configPath, String groupId) throws Exception {
        Properties props = new Properties();
        KafkaProperties kafkaProc = KafkaProperties.getInstance(configPath);

        // Broker连接地址
        props.put(BOOTSTRAP_SERVER, kafkaProc.getValues(BOOTSTRAP_SERVER, "localhost:21007"));
        // Group id
        //props.put(GROUP_ID, kafkaProc.getValues(GROUP_ID, "DemoConsumer"));
        props.put(GROUP_ID, groupId);
        // 是否自动提交offset
        props.put(ENABLE_AUTO_COMMIT, kafkaProc.getValues(ENABLE_AUTO_COMMIT, "true"));
        // 自动提交offset的时间间隔
        props.put(AUTO_COMMIT_INTERVAL_MS, kafkaProc.getValues(AUTO_COMMIT_INTERVAL_MS,"1000"));
        // 会话超时时间
        props.put(SESSION_TIMEOUT_MS, kafkaProc.getValues(SESSION_TIMEOUT_MS, "30000"));
        // 消息Key值使用的反序列化类
        props.put(KEY_DESERIALIZER,
                kafkaProc.getValues(KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer"));
        // 消息内容使用的反序列化类
        props.put(VALUE_DESERIALIZER,
                kafkaProc.getValues(VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer"));
        // 安全协议类型
        props.put(SECURITY_PROTOCOL, kafkaProc.getValues(SECURITY_PROTOCOL, "SASL_PLAINTEXT"));
        // 服务名
        props.put(SASL_KERBEROS_SERVICE_NAME, "kafka");
        // 域名
        props.put(KERBEROS_DOMAIN_NAME, kafkaProc.getValues(KERBEROS_DOMAIN_NAME, "hadoop.hadoop.com"));

        return props;
    }
}
